%%%-------------------------------------------------------------------
%%% File    : egraph.erl
%%% Author  :  vjache
%%% Description : 
%%%
%%% Created : 29 Jan 2010 by  vjache
%%%-------------------------------------------------------------------
-module(erete).

%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include_lib("eunit/include/eunit.hrl").
-include("../../include/erete.hrl").
-include("../../include/pipe.hrl").
-include("../../include/iter.hrl").

%%-------------------------------------------------------------------
%% API export
%%-------------------------------------------------------------------
-export([new/1]).

-record(erete,{name,graph}).
-record(red_node,{red_fun}).
-record(input_node,{iter,period}).
-record(output_node,{pipe}).
-record(mapper_flow,{map_fun}).

%%====================================================================
%% API
%%====================================================================

new(Name)->
	#erete{
		   name=Name,
		   graph=digraph:new([acyclic])
		  }.

%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add_reducer(E,RedFun) when 
  is_record(E,erete), is_function(RedFun,3) ->
	G=E#erete.graph,
	Id=digraph:add_vertex(G),
	digraph:add_vertex(G,Id,#red_node{red_fun=RedFun}).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add_input(E,Iter) when 
  is_record(E,erete), ?IS_ITER(Iter) ->
	G=E#erete.graph,
	Id=digraph:add_vertex(G),
	digraph:add_vertex(G,Id,#input_node{iter=Iter,period=infinity}).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add_input(E,Iter,Period) when 
  is_record(E,erete), 
  ?IS_ITER(Iter),
  is_integer(Period),
  Period>0 ->
	G=E#erete.graph,
	Id=digraph:add_vertex(G),
	digraph:add_vertex(G,Id,#input_node{iter=Iter,period=Period}).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add_output(E,Pipe) when 
  is_record(E,erete), ?IS_PIPE(Pipe) ->
	G=E#erete.graph,
	Id=digraph:add_vertex(G),
	digraph:add_vertex(G,Id,#output_node{pipe=Pipe}).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add_flow(E,S,D,MapFun) when 
  is_record(E,erete),
  is_function(MapFun,1) ->
	G=E#erete.graph,
	case digraph:vertex(G, S) of
		false ->
			throw({no_src_node,S});
		{S,SL} when is_record(SL,red_node) orelse 
					is_record(SL,input_node) ->
			ok;
		BadSrcNode ->
			throw({bad_src_node,BadSrcNode})
	end,
	case digraph:vertex(G, D) of
		false ->
			throw({no_dst_node,S});
		{D,DL} when is_record(DL,red_node) ->
			ok;
		BadDstNode ->
			throw({bad_dst_node,BadDstNode})
	end,
	digraph:add_edge(G, S, D, #mapper_flow{map_fun=MapFun}).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add_flow(E,S,D) when 
  is_record(E,erete) ->
	G=E#erete.graph,
	case digraph:vertex(G, S) of
		false ->
			throw({no_src_node,S});
		{S,SL} when is_record(SL,red_node) orelse 
					is_record(SL,input_node) ->
			ok;
		BadSrcNode ->
			throw({bad_src_node,BadSrcNode})
	end,
	case digraph:vertex(G, D) of
		false ->
			throw({no_dst_node,S});
		{D,DL} when is_record(DL,output_node) ->
			ok;
		BadDstNode ->
			throw({bad_dst_node,BadDstNode})
	end,
	digraph:add_edge(G, S, D, output_flow).
validate(E) when 
  is_record(E,erete) ->
	G=E#erete.graph,
	% 1. 
	VFun=fun({V,R}) when is_record(R,red_node)->
				 InDeg=digraph:in_degree(G, V),
				 OutDeg=digraph:out_degree(G, V),
				 if 
					 InDeg>0,OutDeg>0 ->
						 ok;
					 true ->
						 {bad_reducer,{in_deg,InDeg,out_deg,OutDeg}}
				 end;
			({V,In}) when is_record(In,input_node)->
				 InDeg=digraph:in_degree(G, V),
				 OutDeg=digraph:out_degree(G, V),
				 if 
					 InDeg==0,OutDeg>0 ->
						 ok;
					 true ->
						 {bad_input,{in_deg,InDeg,out_deg,OutDeg}}
				 end;
			({V,Out}) when is_record(Out,output_node)->
				 InDeg=digraph:in_degree(G, V),
				 OutDeg=digraph:out_degree(G, V),
				 if 
					 InDeg>0,OutDeg==0 ->
						 ok;
					 true ->
						 {bad_output,{in_deg,InDeg,out_deg,OutDeg}}
				 end;
			(UexpV) ->
				 {unexpected_node,UexpV}
		 end,
	lists:foldl(
	  fun(V,Prblems)->
		 VL=digraph:vertex(G, V),
		 case VFun(VL) of
			 ok ->
				 Prblems;
			 P ->
				 [{VL,P}|Prblems]
		 end
	  end, [], digraph:vertices(G)).
%%====================================================================
%% Internal functions
%%====================================================================
basic_test() ->
	E=new(test),
	I1=add_input(E,iter:empty()),
	I2=add_input(E,iter:empty()),
	R=add_reducer(E,fun(_MdfCode,_Key,IterList)-> iter_join_cprod:all(IterList) end),
	add_flow(E,I1,R,fun({K,V})-> {K,{K,V}} end),
 	add_flow(E,I2,R,fun({K,V})-> {K,{K,V}} end),
	O=add_output(E,pipe_console:new()),
	add_flow(E,R,O),
	validate(E).
